package com.shujia.flink.core;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Demo01WordCount {
    public static void main(String[] args) throws Exception {
        // 创建Flink的环境：不需要像Spark一样指定master（local[*]），flink任务能够自动识别运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//        env.setParallelism(2);

        // 设置任务上下游发送数据批次延时，默认是100ms
        env.setBufferTimeout(100);

        // 开发flink程序构建环境之后需要得到一个DataStream类型的对象
        // DataStream即Flink中的编程模型

        // 数据会以轮询的方式从nc到flink任务的每个线程中
        DataStream<String> lineDS = env.socketTextStream("master", 8888);

        // 使用print可以将流中的数据打印
        //        lineDS.print();

        // 将每个单词切分并将每个单词变成一行
        DataStream<String> wordsDS = lineDS.flatMap(new MyFlatMapFunction());

        // 将每个单词变成(word,1)的形式
        // 使用匿名内部类
        DataStream<Tuple2<String, Integer>> wordKVDS = wordsDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            /**
             *
             * @param word 输入进来的每个单词
             * @return 返回一个二元组
             */
            @Override
            public Tuple2<String, Integer> map(String word) throws Exception {
                return Tuple2.of(word, 1);
            }
        });


        // 按每个单词进行分组
        // 使用lambda表达式
        KeyedStream<Tuple2<String, Integer>, String> keyByDS = wordKVDS.keyBy(kv -> kv.f0);
        // keyBy之后可以进行聚合操作，聚合操作是KeyedStream类型特有的方法，即：需要先KeyBy才能做聚合
        DataStream<Tuple2<String, Integer>> wordCntDS = keyByDS.sum(1); // sum(1)表示按照位置为1的列进行sum统计

        wordCntDS.print();

        // 启动流处理任务
        env.execute();


    }
}

// 自定义类实现方法的接口

/**
 * 需要指定两个泛型：T、O
 * T：输入的DS中每条数据的类型
 * O：输出到下游中每条数据的类型
 */
class MyFlatMapFunction implements FlatMapFunction<String, String> {

    /**
     * @param line 输入的一条数据
     * @param out  用于将数据发送到下游处理
     */
    @Override
    public void flatMap(String line, Collector<String> out) throws Exception {

        for (String word : line.split(",")) {
            // out每执行一次就会像下游发送一条数据
            out.collect(word);
        }

    }
}
